#ifndef _VL_RTSP_QUEUE_THREAD__H_
#define _VL_RTSP_QUEUE_THREAD__H_

#include <queue>
#include <mutex>
#include <memory>
#include <thread>
#include "defs.h"

namespace VLRTSP {

/**
 * @Author: CimZzz
 * @Date: 2022-01-07 17:34:30
 * 
 * 停止状态（初始状态）
 */
#define _QUEUE_THREAD_STATUS_STOPPED 0

/**
 * @Author: CimZzz
 * @Date: 2022-01-07 17:34:24
 * 
 * 运行中状态
 */
#define _QUEUE_THREAD_STATUS_RUNNING 1


/**
 * @Author: CimZzz
 * @Date: 2022-01-07 17:34:42
 * 
 * 已销毁状态
 */
#define _QUEUE_THREAD_STATUS_DESTROY 2

/**
 * @Author: CimZzz
 * @Date: 2022-01-07 17:13:45
 * 
 * 消息队列线程
 */
template<typename T>
class QueueThread {
public:
	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-14 18:31:32
	 * 
	 * 可执行回调
	 */
	using Runnable = std::function<void()>;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:30:15
	 * 
	 * 消息处理回调方法
	 */
	using MessageCallback = std::function<void(std::shared_ptr<T>)>;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-11 17:07:57
	 * 
	 * 消息生产回调
	 */
	using MessageProducer = std::function<bool(std::shared_ptr<T>&)>;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:30:36
	 * 
	 * 构造函数，需指定消息处理回调
	 */
	explicit QueueThread(MessageCallback handler)  {
		this->status = _QUEUE_THREAD_STATUS_STOPPED;
		this->handler = handler;
		this->queueThread = nullptr;
	}
	
	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:31:05
	 * 
	 * 析构函数，需要回收全部资源，并且关闭线程
	 */
	~QueueThread() {
		this->destroy();
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:31:25
	 * 
	 * 启动消息队列线程
	 */
	void start() {
		std::unique_lock<std::mutex> lock(this->threadMutex);
		if(this->status == _QUEUE_THREAD_STATUS_DESTROY || this->status == _QUEUE_THREAD_STATUS_RUNNING) {
			return;
		}
		this->status = _QUEUE_THREAD_STATUS_RUNNING;
		this->queueThread = new std::thread(&QueueThread::_threadLooper, this);
		this->queueThread->detach();
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-11 17:09:06
	 * 
	 * 添加消息生产线程
	 */
	void addProducer(MessageProducer producer) {
		std::unique_lock<std::mutex> lock(this->threadMutex);
		if(this->status == _QUEUE_THREAD_STATUS_DESTROY) {
			return;
		}

		std::thread* producerThread = new std::thread(&QueueThread::_producerLooper, this, producer);
		producerThread->detach();
		std::shared_ptr<std::thread> ptr(producerThread);
		this->producerVector.push_back(ptr);
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:31:32
	 * 
	 * 向消息队列线程中发送数据
	 * @param data 数据指针
	 */
	void send(T* data) {
		if(data == nullptr) {
			return;
		}
		std::unique_lock<std::mutex> lock(this->threadMutex);
		if(this->status == _QUEUE_THREAD_STATUS_DESTROY) {
			delete data;
			return;
		}
		this->dataQueue.push(std::shared_ptr<QueueMessage>(new QueueMessage(std::shared_ptr<T>(data), nullptr)));
		this->condition.notify_all();
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:31:32
	 * 
	 * 向消息队列线程中发送数据
	 * @param data 数据智能指针
	 */
	void send(std::shared_ptr<T> data)  {
		std::unique_lock<std::mutex> lock(this->threadMutex);
		if(this->status == _QUEUE_THREAD_STATUS_DESTROY) {
			return;
		}
		this->dataQueue.push(std::shared_ptr<QueueMessage>(new QueueMessage(data, nullptr)));
		this->condition.notify_all();
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-14 18:38:58
	 * 
	 * 向消息队列线程中发送可执行回调
	 * @param runnable 可执行回调
	 */
	void sendRunnable(Runnable runnable) {
		std::unique_lock<std::mutex> lock(this->threadMutex);
		if(this->status == _QUEUE_THREAD_STATUS_DESTROY) {
			return;
		}
		this->dataQueue.push(std::shared_ptr<QueueMessage>(new QueueMessage(std::shared_ptr<T>(nullptr), runnable)));
		this->condition.notify_all();
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-10 11:11:59
	 * 
	 * 销毁当前消息队列
	 */
	void destroy()  {
		this->threadMutex.lock();
		if(this->status == _QUEUE_THREAD_STATUS_DESTROY) {
			this->threadMutex.unlock();
			return;
		}
		this->status = _QUEUE_THREAD_STATUS_DESTROY;
		std::thread* threadPtr = this->queueThread;
		this->queueThread = nullptr;
		this->threadMutex.unlock();
		if(threadPtr != nullptr) {
			// 释放队列锁
			this->condition.notify_all();
			// 等待线程结束
			try {
				if(threadPtr->joinable()) {
					threadPtr->join();
				}
			}
			catch(std::system_error error) {
				// join error, if thread is done, maybe throw this error, ignore
			}
			
			delete threadPtr;
		}
		for(auto iter = this->producerVector.begin() ; iter != this->producerVector.end() ; iter ++) {
			try {
				std::thread* threadPtr = (*iter).get();
				if(threadPtr->joinable()) {
					threadPtr->join();
				}
			}
			catch(std::system_error error) {
				// join error, if thread is done, maybe throw this error, ignore
			}
		}
	}

private:
	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-14 18:28:51
	 * 
	 * 消息队列消息
	 */
	struct QueueMessage {
		QueueMessage(std::shared_ptr<T> dataPtr, Runnable runnable): dataPtr(dataPtr) {
			this->runnable = runnable;
		}
		std::shared_ptr<T> dataPtr;
		Runnable runnable;
	};

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-11 17:11:56
	 * 
	 * 生产者消息队列
	 */
	void _producerLooper(MessageProducer producer) {
		while(this->status == _QUEUE_THREAD_STATUS_RUNNING) {
			std::shared_ptr<T> ptr(nullptr);
			if(!producer(ptr)) {
				break;
			}
			if(ptr.get() != nullptr) {
				this->send(ptr);
			}
		}
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 18:16:15
	 * 
	 * 线程循环
	 */
	void _threadLooper()  {
		while(this->status == _QUEUE_THREAD_STATUS_RUNNING) {
			// 从 Vector 中取一个数据
			std::shared_ptr<QueueMessage> message = this->_checkQueue();
			QueueMessage* messagePtr = message.get();
			if(messagePtr == nullptr) {
				continue;
			}
			T* dataPtr = messagePtr->dataPtr.get();
			if(dataPtr != nullptr) {
				this->handler(messagePtr->dataPtr);
			} else if(messagePtr->runnable) {
				messagePtr->runnable();
			}
		}
	}

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 19:08:43
	 * 
	 * 检查消息队列是否有足够数据进行循环，如果数据不足
	 * 则会触发条件变量直到解锁，并返回消息队列头部数据
	 * @return 数据智能指针
	 */
	std::shared_ptr<QueueMessage> _checkQueue()  {
		std::unique_lock<std::mutex> uniqueLock(this->threadMutex);
		
		if(this->dataQueue.empty()) {
			this->condition.wait(uniqueLock);
		}
		if(!this->dataQueue.empty()) {
			std::shared_ptr<QueueMessage> dataPtr = this->dataQueue.front();
			this->dataQueue.pop();
			return dataPtr;
		}

		return std::shared_ptr<QueueMessage>(nullptr);
	}



	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:35:09
	 * 
	 * 当前消息队列线程状态
	 */
	volatile uint8_t status;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:35:51
	 * 
	 * 消息处理回调方法
	 */
	MessageCallback handler;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:32:36
	 * 
	 * 线程锁
	 */
	std::mutex threadMutex;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 18:17:40
	 * 
	 * 条件变量
	 */
	std::condition_variable condition;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:32:41
	 * 
	 * 内部线程
	 */
	std::thread* queueThread;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-11 17:10:07
	 * 
	 * 生产者线程列表
	 */
	std::vector< std::shared_ptr<std::thread> > producerVector;

	/**
	 * @Author: CimZzz
	 * @Date: 2022-01-07 17:32:53
	 * 
	 * 消息队列容器
	 */
	std::queue< std::shared_ptr<QueueMessage> > dataQueue;
};

}
#endif //_VL_RTSP_QUEUE_THREAD__H_
